package com.lagou.rpc.consumer;

import com.lagou.rpc.consumer.client.RpcClient;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class ZookeeperClient {


    public static void main(String[] args) throws Exception {
        //建立zk连接
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                //zk的server地址，多个server之间使⽤英⽂逗号分隔开
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();

        //获取所有节点列表
        String parentPath = "/zk";
        List<String> nodeList = zkClient.getChildren().forPath(parentPath);
        if (CollectionUtils.isEmpty(nodeList)) {
            throw new RuntimeException("连接失败，获取不到节点列表");
        }
        for (String s : nodeList) {
            String nodePath = parentPath + "/" + s;
            //判断节点的状态是否在线
            Stat stat = zkClient.checkExists().forPath(nodePath + "/stat");
            if (stat == null) {
                //不在线直接跳过
                continue;
            }
            //获取节点里的服务端数据
            byte[] bytes = zkClient.getData().forPath(nodePath);
            String data = new String(bytes, StandardCharsets.UTF_8);
            String[] split = data.split(":");
            String ip = split[0];
            String port = split[1];
            RpcClient rpcClient = new RpcClient(ip, Integer.parseInt(port));
            System.out.println(data + "建立连接成功");
            //对状态节点进行监听 dataIsCompressed:是否进行数据压缩
            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, nodePath, false);
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    if (event.getData().getPath().equals(nodePath + "/stat") && event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                        Thread.sleep(1000);
                        System.out.println(data + "重新建立连接成功");
                        RpcClient rpcClient = new RpcClient(ip, Integer.parseInt(port));
                    }
                    if (event.getData().getPath().equals(nodePath + "/stat") && event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                        System.out.println("删除节点: " + data + "监听成功");
                        rpcClient.close();
                    }
                }
            });
            Thread.sleep(Integer.MAX_VALUE);
            /*zkClient.checkExists().usingWatcher(new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    System.out.println(watchedEvent);
                    if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
                        System.out.println("删除节点: " + data + "监听成功");
                        rpcClient.close();
                    }
                    if (watchedEvent.getType() == Event.EventType.NodeCreated) {
                        System.out.println(data + "重新建立连接成功");
                        RpcClient rpcClient = new RpcClient(ip, Integer.parseInt(port));
                    }
                }
            }).forPath(nodePath + "/stat");
            Thread.sleep(Integer.MAX_VALUE);*/
        }
    }
}
